package com.business.etl.flow.listener;

import cn.hutool.core.util.StrUtil;
import com.component.api.constants.FlowStatus;
import com.component.api.constants.StatusContents;
import com.common.utils.constant.Constants;
import com.business.etl.flow.service.FlowService;
import com.framework.redis.service.RedisService;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.connection.RedisConnection;
import org.springframework.data.redis.listener.PatternTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.util.StringUtils;

import java.nio.charset.Charset;
import java.util.Properties;

/**
 * 描述：
 * redis key值变化监听
 *
 * @author xianggj
 * @Date 2021/12/13 11:38
 **/
public class RedisKeyChangeListener implements MessageListener {

    private RedisService redisService;

    private FlowService flowService;

    /**
     * 监听事件 K表示键空间 E表示键事件 h表示只处理hash key  A表示处理所有
     */
    private String keyspaceNotificationsConfigParameter = "KE$";

//    private static final Topic TOPIC_ALL_KEYEVENTS = new PatternTopic("__keyevent@*"); //表示只监听所有的key
    /**
     * 表示只监听所有的key的hset事件
     */
//    private static final Topic TOPIC_KEYEVENTS_SET = new PatternTopic("__keyevent@0__:hset");
    /**
     * 指定key的所有事件 指定wj12:flow开头的
     */
    private static final Topic TOPIC_KEYSPACE_SET = new PatternTopic("__keyspace@0__:" + StatusContents.FLOW_STATUS_KEY + "*");


    public RedisKeyChangeListener(RedisMessageListenerContainer listenerContainer,
                                  RedisService redisService, FlowService flowService) {
        this.redisService = redisService;
        this.flowService = flowService;
        initAndSetRedisConfig(listenerContainer);
    }

    public void initAndSetRedisConfig(RedisMessageListenerContainer listenerContainer) {
        if (StringUtils.hasText(keyspaceNotificationsConfigParameter)) {
            RedisConnection connection = listenerContainer.getConnectionFactory().getConnection();
            try {
                Properties config = connection.getConfig("notify-keyspace-events");
                if (!StringUtils.hasText(config.getProperty("notify-keyspace-events"))) {
                    connection.setConfig("notify-keyspace-events", keyspaceNotificationsConfigParameter);
                }
            } finally {
                connection.close();
            }
        }
        // 注册消息监听
        listenerContainer.addMessageListener(this, TOPIC_KEYSPACE_SET);
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String key = new String(message.getChannel(), Charset.forName("UTF-8"));
        String jobId = StrUtil.subAfter(key, Constants.COLON, true);
        String status = redisService.get(StatusContents.FLOW_STATUS_KEY + jobId);
        if (status != null){
            if (FlowStatus.STOP.name().equalsIgnoreCase(status)){
                flowService.removeKey(jobId);
            }
        }
    }
}
